package com.atguigu.gmall.apps

import java.lang
import java.time.{Instant, LocalDateTime, ZoneId}
import java.time.format.DateTimeFormatter

import com.alibaba.fastjson.JSON
import com.atguigu.gmall.Beans.StartLog
import com.atguigu.gmall.utils.Constant.KafkaConstant
import com.atguigu.gmall.utils.{MykafkaUtil, RedisUtil}
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.spark.streaming.dstream.DStream
import org.slf4j.{Logger, LoggerFactory}
import org.apache.phoenix.spark._
//使用了隐式转换，可以直接使用rdd调用phoenix的方法

/**
  * @ClassName: DAUApp2
  * @Description:
  * @Author: kele
  * @Date: 2021/3/19 12:28
  **/
object DAUApp extends baseApp {

  override val AppName: String = "DAU"


  override val time: Int = 10
  def main(args: Array[String]): Unit = {

    //控制抽象，直接获取ssc的连接
    runApp {
      //1、获取kafka中的启动日志
      val ds = MykafkaUtil.getKafkaStream(KafkaConstant.KAFKA_TOPIC_STARTUP, streamingContext)


      /**获取的数据为
        * {"area":"beijing","uid":"222","os":"andriod","ch":"tencent",
        * "appid":"gmall2019","mid":"mid_472",
        * "type":"startup","vs":"1.1.3","ts":1615951376195}
        */



      //2、将字符串转化为json对象
      var ds1 = ds.map(startLog => {

        //获得StartLog的对象
        val log = JSON.parseObject(startLog.value(), classOf[StartLog])

        log.logDate =  LocalDateTime.ofInstant(Instant.ofEpochMilli(log.ts),ZoneId.of("Asia/Shanghai"))
          .format(DateTimeFormatter.ofPattern("yyyy-MM-dd"))

        log.logHour = LocalDateTime.ofInstant(Instant.ofEpochMilli(log.ts),ZoneId.of("Asia/Shanghai"))
          .format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH")).split(" ")(1)

        log
      })

      ds1.count().print()

      //3、处理json对象，判断是否重复（分区内判断和分区间判断数据）

      val ds2 = RemoveDuplicates(ds1)

      ds2.count().print()

      val ds3 = RemoveDuplicatesBatch(ds2)

      //如果之后有多个状态需要使用ds3，且后面使用ds3时需要多次处理，则ds3需要保存状态,这里可以不需要设置
      val ds4 = ds3.cache()

      ds3.count().print()

      //4、将数据写入redis中
      WriteToRedis(ds4)

      //5、将数据写入Hbase中
      WriteToHBase(ds4)

    }


    /**
      * 分区内去重
      *
      * 思路：
      * 判断 传入的每条数据的mid 是否 在redis中当天key下的set中的mid中，
      * 细节：
      * 1、从redis池中获取连接，使用map（获取连接的次数=数据的个数）还是mappartitions（获取的连接次数=分区数）
      *
      * @param ds
      * @return
      */
    def RemoveDuplicates(ds: DStream[StartLog]) = {

      //采用mappartitions的方式，减少连接次数
      ds.mapPartitions(rdd => {
        //1、在分区后获取连接，注意不能在之前获取连接
        val jedisClient = RedisUtil.getJedisClient

        //2、过滤掉已经存在的mid
        var ds1 = rdd.filter(startlog => {

          //存在应该返回0，不存在返回1
          !jedisClient.sismember("DAU" + startlog.logDate, startlog.mid)
        })

        //3、关闭连接
        jedisClient.close()

        //返回ds
        ds1

      })
    }

    /**
      * 分区间去重
      * 思路：
      * 不能使用filter：filter需要数据对比，每个分区的数据都需要进行判断，没有对比的数据
      *
      * 使用group的方式对各个分区的数据进行聚合
      * 细节：
      * 1、由于是10s一个批次，存在跨天的去重，所以需要按照ts和mid作为key进行去重
      *
      * @param ds
      */
    def RemoveDuplicatesBatch(ds: DStream[StartLog]) = {

      //使用transform，有返回值，且返回值的类型为ds
      //foreachRdd没有返回值
      val ds3 = ds.transform(rdd => {

        //1、设置数据格式，key为（mid，date）
        rdd.map(startlog => {
          //特定方式的数据（（mid，date），startlog）
          ((startlog.mid, startlog.logDate), startlog)
        })
          //2、按照key进行去重，会将重复的startlog作为list，只需要保存第一次启动的日志
            .groupByKey()
          //3、对启动日志进行排序，保留第一次的启动时的日志
            .flatMap({

          case ((mid, date), log) => {
            log.toList.sortBy(_.ts).take(1)
          }
        })
      })
      //将数据启动日志的对象返回
      ds3

    }


    /**
      * 将数据写入Redis中
      *
      * 思路：
      * 获取连接，将每条数据写入
      *
      * 细节：
      * 1、不需要返回值，使用foreachrdd
      *
      * @param ds3
      * @return
      */

    def WriteToRedis(ds3: DStream[StartLog]) = {

      //1、将DStreaming转化为RDD
      ds3.foreachRDD(rdd => {
        //2、对每个分区进行批次处理
        rdd.foreachPartition(rdd1 => {

          //val logger = LoggerFactory.getLogger("logger")
          //3、获取连接
          val jedisClient = RedisUtil.getJedisClient


          /**
            *  map将数据插入不进去，需要使用foreach
            *
            *原因： 使用map连接不到redis,(map之后算子必须再使用才可以调用该方法)
            *
            * 例如： val longs: Iterator[lang.Long] = rdd1.map(......)
            *       longs.size
            *       这样map中的就可以执行了,使用foreachrdd可以不用定义变量名
            *
            */

          rdd1.foreach(startlog => {
            //4、将数据写入到redis中
            // logger.error(startlog.toString);
            jedisClient.sadd("DAU" + startlog.logDate, startlog.mid)
          })

          //longs.size

          //logger.error(longs.size.toString)

          //5、关闭连接
          jedisClient.close()
        })
      })
    }

    /**
      * 将数据保存在HBase上
      * @param ds
      */
    def WriteToHBase(ds: DStream[StartLog]): Unit = {

      ds.foreachRDD(rdd=>{



        //Phoenix中所有的表名默认是大写的
        rdd.saveToPhoenix("gmall2021_dau".toUpperCase,
          //列名在序列中
          Seq("MID", "UID", "APPID", "AREA", "OS", "CH", "TYPE", "VS", "LOGDATE", "LOGHOUR", "TS"),

          //是hadoop类下的configuration类型，如果直接new，只能读取hadoop下的配置文件，
          //通过hbaseconfiguration可以读取hbase下的配置文件信息
          //HBaseConfiguration.create()： 不仅读取hadoop的配置文件，还会读取hbase-site.xml和hbase-default.xml
          HBaseConfiguration.create(),
          //zk的连接，option类型
          Some("hadoop102:2181")
        )


      })
    }


  }
}
